package com.yihaodian.cluster.admin;

import java.util.Collection;
import java.util.Map;

import org.apache.log4j.Logger;

import com.google.common.collect.Maps;
import com.google.inject.Inject;
import com.yihaodian.api.search.SearchException;
import com.yihaodian.cluster.ClusterService;
import com.yihaodian.cluster.ClusterState;
import com.yihaodian.cluster.ClusterStateUpdate;
import com.yihaodian.cluster.ClusterStateUpdateType;
import com.yihaodian.cluster.Node;
import com.yihaodian.cluster.Node.Role;
import com.yihaodian.cluster.ShardMgtChange;
import com.yihaodian.cluster.discovery.zk.ZkNodeDiscovery;
import com.yihaodian.index.meta.IndexMeta;
import com.yihaodian.index.meta.routing.RoutingTable;
import com.yihaodian.index.meta.shard.Shard;
import com.yihaodian.index.meta.shard.ShardId;
import com.yihaodian.index.meta.shard.ShardRouting;
import com.yihaodian.server.NodeBootstrap;

/**
 * Master Node
 */
public class ClusterAdminService implements ClusterAdmin {
	private static final Logger logger = Logger
			.getLogger(ClusterAdminService.class);

	ClusterService clusterService;

	@Inject
	public ClusterAdminService(ClusterService clusterService) {
		this.clusterService = clusterService;
	}

	@Override
	public void setRoutingTable(RoutingTable routingTable) {
		ensureMaster();

		ClusterStateUpdate update = new ClusterStateUpdate(
				ClusterStateUpdateType.cluster_routing_table_change);
		update.setRoutingTable(routingTable);
		clusterService.updateClusterState(update);
	}

	@Override
	public void setIndexMeta(IndexMeta meta) {
		ensureMaster();

		ClusterStateUpdate update = new ClusterStateUpdate(
				ClusterStateUpdateType.cluster_index_meta_change);
		update.setIndexMeta(meta);
		clusterService.updateClusterState(update);
	}

	@Override
	public void setNodeShardStatus(Node node, ShardRouting shard) {
		// TODO Auto-generated method stub

	}

	@Override
	public void disableShard(ShardId shard, boolean allNode, Node targetNode) {
		ensureMaster();
		logger.info("disable shard: " + shard + " " + allNode + targetNode != null ? ""
				: targetNode.getId());

		ClusterStateUpdate update = new ClusterStateUpdate(
				ClusterStateUpdateType.shard_mgt_change);

		ShardMgtChange change = new ShardMgtChange(shard, false, allNode,
				targetNode);
		update.setShardMgtChange(change);

		clusterService.updateClusterState(update);
	}

	@Override
	public void enableShard(ShardId shard, boolean allNode, Node targetNode) {
		ensureMaster();
		logger.info("enable shard: " + shard + " " + allNode
				+ (targetNode == null ? "" : targetNode.getId()));

		ClusterStateUpdate update = new ClusterStateUpdate(
				ClusterStateUpdateType.shard_mgt_change);

		ShardMgtChange change = new ShardMgtChange(shard, true, allNode,
				targetNode);
		update.setShardMgtChange(change);

		clusterService.updateClusterState(update);
	}

	@Override
	public void addShard(ShardId shard, Node targetNode) {
		ensureMaster();
		if (targetNode == null) {
			throw new SearchException("targetNode can not be null");
		}
		logger.info("add shard: " + shard + " " + targetNode.getId());
		ClusterStateUpdate update = new ClusterStateUpdate(
				ClusterStateUpdateType.cluster_routing_table_change);
		ClusterState cs = clusterService.state();
		ShardRouting newShardRouting = new ShardRouting(shard);
		RoutingTable rt = cs.routingTables().routingTable(shard.getIndexId())
				.clone();
		rt.addEntry(targetNode.getId(), newShardRouting);
		update.setRoutingTable(rt);
		clusterService.updateClusterState(update);

	}

	@Override
	public void removeShard(ShardId shard, Node targetNode) {
		ensureMaster();
		if (targetNode == null) {
			throw new SearchException("targetNode can not be null");
		}
		logger.info("remove shard: " + shard + " " + targetNode.getId());
		ClusterStateUpdate update = new ClusterStateUpdate(
				ClusterStateUpdateType.cluster_routing_table_change);
		ClusterState cs = clusterService.state();
		RoutingTable rt = cs.routingTables().routingTable(shard.getIndexId())
				.clone();
		rt.removeEntry(targetNode.getId(), shard);
		update.setRoutingTable(rt);
		clusterService.updateClusterState(update);
	}

	@Override
	public void setShardDataVersion(ShardId shardId, String version) {
		ClusterStateUpdate update = new ClusterStateUpdate(
				ClusterStateUpdateType.shard_data_version_change);

		Map<ShardId, String> dataVersion = Maps.newHashMap();
		dataVersion.put(shardId, version);
		update.setDataVersion(dataVersion);

		clusterService.updateClusterState(update);

	}

	void ensureMaster() {
		boolean isMaster = clusterService.getLocalNode().getRole() == Role.MASTER;
		if (!isMaster)
			throw new SearchException(
					"not master, can not execute CluseterAdmin");
	}

	public static void main(String[] args) throws InterruptedException {
		ZkNodeDiscovery.clearState = true;
		NodeBootstrap nodeLanuncher = new NodeBootstrap(8180);
		nodeLanuncher.getNode().setRole(Role.MASTER);
		nodeLanuncher.startup();

		ClusterAdmin clusterAdmin = nodeLanuncher.getInjector().getInstance(
				ClusterAdmin.class);

		Thread.sleep(15000);

		String[] indices = { "product", "code" };
		for (String index : indices) {
			IndexMeta meta = new IndexMeta(index, 2);

			if (index.equals("code"))
				meta.setShardHandlerParam("projectName");
			clusterAdmin.setIndexMeta(meta);

			RoutingTable routingTable = new RoutingTable(index);
			Collection<Shard> shards = meta.getShards();

			Node[] nodes = new Node[] { new Node(8180), new Node(8181) };

			for (Shard shard : shards) {
				for (Node node : nodes) {
					routingTable.addEntry(node.getId(),
							new ShardRouting(shard.getShardId()));
				}
			}
			clusterAdmin.setRoutingTable(routingTable);

			Thread.sleep(5 * 1000);
		}

		nodeLanuncher.shutdown();
	}

}
